package com.chukun.flink.stream.window.process.function;

import com.chukun.flink.stream.window.process.accumulator.AverageAccumulator;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple3;

/**
 * @author chukun
 * @version 1.0.0
 * @description 平均值的聚合函数
 * @createTime 2022年05月23日 00:00:00
 */
public class AverageAggregation implements AggregateFunction<Tuple3<String, Integer, String>, AverageAccumulator, AverageAccumulator> {
    @Override
    public AverageAccumulator createAccumulator() {
        return new AverageAccumulator();
    }

    @Override
    public AverageAccumulator add(Tuple3<String, Integer, String> tuple, AverageAccumulator acc) {
        acc.setWord(tuple.f0);
        acc.setCount(acc.getCount() + 1);
        acc.setSum(acc.getSum() + tuple.f1);
        return acc;
    }

    @Override
    public AverageAccumulator getResult(AverageAccumulator acc) {
        return acc;
    }

    @Override
    public AverageAccumulator merge(AverageAccumulator acc01, AverageAccumulator acc02) {
        acc01.setCount(acc01.getCount() + acc02.getCount());
        acc01.setSum(acc01.getSum() + acc02.getSum());
        return acc01;
    }
}
